package function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.traffic.common.utils.JdbcUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

public class TrafficLogClickHouse {
    public static void saveTrafficStatus(DataStream<String> stream) {
        /*
            {
              "traffic_status": {
                "road": "G0422武深高速株洲段（醴娄高速）",
                "congestion_level": "畅通",
                "congestion_mileage": 20.5,
                "time": 1748230075194,
                "source": "湖南高速路网监测中心",
                "speed": 74.8
              }
            }
        */
        // 1数据转换
        SingleOutputStreamOperator<String> jsonStream = stream.map(
                new RichMapFunction<String, String>() {
                    @Override
                    public String map(String value) throws Exception {
                        // 1解析JSON
                        JSONObject object = JSON.parseObject(value);
                        JSONObject trafficStatusObject = object.getJSONObject("traffic_status");

                        // 2提取字段值
                        JSONObject jsonObject = new JSONObject();

                        jsonObject.put("road", trafficStatusObject.getString("road"));
                        jsonObject.put("congestion_level", trafficStatusObject.getString("congestion_level"));
                        jsonObject.put("congestion_mileage", trafficStatusObject.getDouble("congestion_mileage"));
                        jsonObject.put("ts", trafficStatusObject.getLong("time"));
                        jsonObject.put("source", trafficStatusObject.getString("source"));
                        jsonObject.put("speed", trafficStatusObject.getDouble("speed"));
                        // 3返回JSON字符串
                        return jsonObject.toJSONString();
                    }
                }
        );

        // 2-1字段名称
        String[] columns = new String[]{
                "road", "congestion_level", "congestion_mileage", "ts", "source", "speed"
        };
        // 2-2插入语句
        String upsertSql = "INSERT INTO traffic_monitoring.dwd_traffic_traffic_status(\n" +
                "    road, congestion_level, congestion_mileage, ts, source, speed\n" +
                ") VALUES (?,?,?,?,?,?)";
        // 2-3插入数据
        JdbcUtil.sinkToClickhouseUpsertJson(jsonStream, columns, upsertSql);
    }


    public static void saveCameras(DataStream<String> stream) {
        SingleOutputStreamOperator<String> jsonStream = stream.map(new RichMapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                JSONObject object = JSON.parseObject(value);
                JSONObject cameraObject = object.getJSONObject("cameras");

                // 提取摄像头字段值
                JSONObject jsonObject = new JSONObject();

                jsonObject.put("camera_id", cameraObject.getString("camera_id"));
                jsonObject.put("location", cameraObject.getString("location"));
                jsonObject.put("status", cameraObject.getString("status"));
                jsonObject.put("video_url", cameraObject.getString("video_url"));
                jsonObject.put("ts", object.getLong("time"));

                return jsonObject.toJSONString();
            }
        });

        // 字段名称
        String[] columns = new String[]{
                "camera_id", "location", "status", "video_url", "ts"
        };

        // 插入语句
        String upsertSql = "INSERT INTO traffic_monitoring.dwd_traffic_cameras(\n" +
                "    camera_id, location, status, video_url, ts\n" +
                ") VALUES (\n" +
                "     ?, ?, ?, ?, ?\n" +
                ")";

        // 插入数据到ClickHouse
        JdbcUtil.sinkToClickhouseUpsertJson(jsonStream, columns, upsertSql);
    }

    public static void saveCongestionRanking(DataStream<String> stream) {
        SingleOutputStreamOperator<String> jsonStream = stream.map(new RichMapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                JSONObject object = JSON.parseObject(value);
                JSONObject rankingObject = object.getJSONObject("congestion_ranking");

                JSONObject jsonObject = new JSONObject();
                jsonObject.put("road", rankingObject.getString("road"));
                jsonObject.put("congestion_mileage", rankingObject.getDouble("congestion_mileage"));
                jsonObject.put("ts", object.getLong("time"));

                return jsonObject.toJSONString();
            }
        });

        String[] columns = new String[]{
                "road", "congestion_mileage", "ts"
        };

        String upsertSql = "INSERT INTO traffic_monitoring.dwd_traffic_congestion_ranking(\n" +
                "    road, congestion_mileage, ts\n" +
                ") VALUES (?, ?, ?)";

        JdbcUtil.sinkToClickhouseUpsertJson(jsonStream, columns, upsertSql);
    }


    public static void saveIncidents(DataStream<String> stream) {
        /*
            {
              "incidents": {
                "camera_id": "CAM_CAM_14",
                "location": "G4京港澳高速长沙段 K71+620M",
                "time": 1748230120407,
                "type": "施工管制",
                "status": "已疏通"
              },
              "time": 1748230120407
            }
        */
        // 1数据转换
        SingleOutputStreamOperator<String> jsonStream = stream.map(
                new RichMapFunction<String, String>() {
                    @Override
                    public String map(String value) throws Exception {
                        // 1解析JSON
                        JSONObject object = JSON.parseObject(value);
                        JSONObject incidentsObject = object.getJSONObject("incidents");

                        // 2提取字段值
                        JSONObject jsonObject = new JSONObject();

                        jsonObject.put("camera_id", incidentsObject.getString("camera_id"));
                        jsonObject.put("location", incidentsObject.getString("location"));
                        jsonObject.put("type", incidentsObject.getString("type"));
                        jsonObject.put("status", incidentsObject.getString("status"));
                        jsonObject.put("ts", object.getLong("time"));
                        // 3返回JSON字符串
                        return jsonObject.toJSONString();
                    }
                }
        );

        // 2-1字段名称
        String[] columns = new String[]{
                "camera_id", "location", "type", "status", "ts"
        };
        // 2-2插入语句
        String upsertSql = "INSERT INTO TABLE traffic_monitoring.dwd_traffic_incidents(\n" +
                "    camera_id, location, type, status, ts\n" +
                ") VALUES (\n" +
                "  ?,?,?,?,?\n" +
                ")";
        // 2-3插入数据
        JdbcUtil.sinkToClickhouseUpsertJson(jsonStream, columns, upsertSql);
    }


    public static void saveTollFlow(DataStream<String> stream) {
        SingleOutputStreamOperator<String> jsonStream = stream.map(new RichMapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                JSONObject object = JSON.parseObject(value);
                JSONObject tollFlowObject = object.getJSONObject("toll_flow");
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("in_flow", tollFlowObject.getString("in_flow"));
                jsonObject.put("out_flow", tollFlowObject.getString("out_flow"));
                jsonObject.put("name", tollFlowObject.getString("name"));
                jsonObject.put("change_rate", tollFlowObject.getDouble("change_rate"));
                jsonObject.put("ts", object.getLong("time"));

                return jsonObject.toJSONString();
            }
        });

        String[] columns = new String[]{
                "in_flow", "out_flow", "name", "change_rate", "ts"
        };


        String upsertSql = "INSERT INTO TABLE traffic_monitoring.dwd_traffic_toll_flow(\n" +
                "    in_flow, out_flow, name, change_rate, ts\n" +
                ") VALUES (\n" +
                "    ?,?,?,?,?\n" +
                ")";

        JdbcUtil.sinkToClickhouseUpsertJson(jsonStream, columns, upsertSql);
    }
}